package com.lry.message;

import com.alibaba.fastjson.annotation.JSONField;
import com.alibaba.fastjson.annotation.JSONType;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import java.beans.Transient;
import java.io.Serializable;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;

@Data
@AllArgsConstructor
@NoArgsConstructor
@Slf4j
public class MessageQueue implements Serializable {

    private String topic;

    private String brokerName;

    private int queueId;

    public static String group = "";

    //<group,offset>
    private volatile Map<String,AtomicLong> consumeOffsetMap = new ConcurrentHashMap<>();

    //延迟消息和未ack的消息统一处理，
    private volatile Map<String, Set<Long>> consumeNoAckOffsetMap = new ConcurrentHashMap<>();//处理未ack的 偏移量

    private volatile Map<String,AtomicLong> produceOffsetMap = new ConcurrentHashMap<>();


    public MessageQueue(String topic, String brokerName, int queueId) {
        this.topic = topic;
        this.brokerName = brokerName;
        this.queueId = queueId;
        produceOffsetMap.put(group,new AtomicLong(0));
    }

    public void incrProduceOffset(){
        for (AtomicLong atomicLong : produceOffsetMap.values()) {
            if(null!=atomicLong){
                atomicLong.incrementAndGet();
            }
        }
    }


    public void removeOffsetByGroup(String group){
        consumeOffsetMap.remove(group);
        produceOffsetMap.remove(group);
    }


    public void addNoAckConsumeOffset(String group,long offset){
        consumeNoAckOffsetMap.get(group).add(offset);
    }

    public void removeNoAckConsumeOffset(String group,long offset){
        consumeNoAckOffsetMap.get(group).remove(offset);
    }

    public List<Long> getConsumeOffset(String group) {
        return getConsumeOffset(group,false);
    }

    public List<Long> getConsumeOffset(String group,boolean autoAck){
        AtomicLong consumeAl = consumeOffsetMap.get(group);
        AtomicLong producerAl = produceOffsetMap.get(group);
        List<Long> list = new ArrayList<>();

        for (Long noAckOffset : consumeNoAckOffsetMap.get(group)) {
            list.add(noAckOffset);
        }

        while(consumeAl.get()<producerAl.get()){
            long consumeOffset = consumeAl.getAndIncrement();
            list.add(consumeOffset);
            if(!autoAck){
                addNoAckConsumeOffset(group,consumeOffset);
            }
        }

        return list;
    }

    @Transient
    public long getProduceOffset(){
        for (AtomicLong atomicLong : produceOffsetMap.values()) {
            return atomicLong.get();
        }
        log.error("getProduceOffset 错误");
        return 0;
    }

    public boolean hasMsg(String group){
        AtomicLong atomicLong = consumeOffsetMap.get(group);
        AtomicLong atomicLong2 = produceOffsetMap.get(group);
        if(null==atomicLong2 || null==atomicLong){
            return false;
        }
        Set<Long> noAck = consumeNoAckOffsetMap.get(group);
        return atomicLong2.get()>atomicLong.get()||
                (noAck!=null && !noAck.isEmpty());
    }

}
